package com.zy.datareceiver.binsess.decode.service;


import com.zy.datareceiver.binsess.decode.util.WarningLevelUtil;
import com.zy.datareceiver.binsess.monitoring.entity.MonitoringData;
import com.zy.datareceiver.binsess.monitoring.entity.MonitoringErrorData;
import com.zy.datareceiver.binsess.monitoring.entity.MonitoringPoint;
import com.zy.datareceiver.binsess.monitoring.mapper.MonitoringErrorDataMapper;
import com.zy.datareceiver.binsess.monitoring.mapper.MonitoringPointMapper;
import com.zy.datareceiver.binsess.monitoring.param.index.DeviceData;
import com.zy.datareceiver.binsess.rabbit.entity.DataGram;
import com.zy.datareceiver.binsess.rabbit.entity.LoopQueue;
import com.zy.datareceiver.binsess.rabbit.entity.Param;
import com.zy.datareceiver.binsess.rabbit.entity.ReceiveDataStorage;
import com.zy.datareceiver.binsess.status.entity.ObjectMonitoring;
import com.zy.datareceiver.binsess.status.mapper.ObjectMonitoringMapper;
import com.zy.datareceiver.binsess.stomp.service.StompMessageService;
import com.zy.datareceiver.binsess.warning.entity.WarningData;
import com.zy.datareceiver.binsess.warning.entity.WarningThreshold;
import com.zy.datareceiver.core.cast.CacheDict;
import com.zy.datareceiver.core.cast.ConCast;
import com.zy.datareceiver.core.common.SpringApplicationContext;
import com.zy.datareceiver.core.utils.DateUtil;
import com.zy.datareceiver.core.utils.JsonUtil;
import com.zy.datareceiver.core.utils.KNN;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;

/**
 * 数据解析核心线程，从队列中取出数据{@link DataGram}
 * step1: 先将数据取出来查看是否是异常数据 ：
 *     1.首先看数据值是否可以转换成正常dedouble类型的数据，详见
 *      @See {@link StorageTaskService#isErrorData(DataGram)
 *     2.数据是正常的double数据之后，取出暂存的数据进行KNN算法判断，查看数据是否是峰值数据，详见
 *      @See {@link StorageTaskService#isExceptionData(Integer, Integer, Double)}
 * step2: 数据是正常的数据之后将接受到的数据的pointId存放入pointIdsQueue队列中，以便后续从
 *      @See {@link StompMessageService#run()}中取出数据并进行前端推送
 * step3: 解析DataGram包中的数据，分别对里面的参数查看预警情况，详见
 *      @See {@link WarningLevelUtil#getLevel(Integer, Integer, Double)}
 *      如果返回的结果中预警等级为-1，则表示是正常数据，其他1、2、3分别表示1、2、3级预警
 *      如果是异常数据，即返回的结果不是-1，则进行预警数据存储，将{@link DataGram} 转换成
 *      {@link WarningData},并存入warningQueue中进行入库操作，详见
 *      @See {@link com.zy.datareceiver.binsess.warning.service.impl.WarningDataTask#run() }
 * step4: 实时数据缓存在本地的队列中，无论是否是异常数据，都应该再实时数据中进行入库，将{@link DataGram}
 *      转换成{@link MonitoringData}形式，存入内存循环队列{@link LoopQueue}中，等待前端推送
 * step5: 实时数据入库，昨完以上的操作之后将{@link DataGram}转换成{@link MonitoringData}存入队列，
 *      等待入库，详见 @See {@link DBSaveThread#run()},至此整个核心的业务流程已经结束
 * @version 1.0
 * @desc : 由于前面单独线程从rabbitmq取出数据之后直接到这里解析，如果这个地方堆积，会影响整体项目的时效性，
 *          需要将参数@See {@link ConCast.Thread#DECODE} 调大，保证下游数据通畅
 * */
public class StorageTaskService extends Thread {

    // <pointName,<parameterNameEn, LoopQueue<DeviceData>>>
    public static final HashMap<String, HashMap<String,LoopQueue<DeviceData>>> pointMap= new HashMap<>(256);

    private static final Logger log = LoggerFactory.getLogger(StorageTaskService.class);

    public static final BlockingQueue<MonitoringData> monitoringDataQueue = new LinkedBlockingDeque<MonitoringData>(ConCast.QueueSize.MONITORING_DATA_QUEUE_SIZE);//实时推送数据集合
    public static final BlockingQueue<WarningData> warningQueue = new LinkedBlockingDeque<WarningData>(ConCast.QueueSize.WARNING_DB_QUEUE_SIZE);//预警数据集合
    public static final BlockingQueue<WarningData> sendMsgQueue = new LinkedBlockingDeque<WarningData>(ConCast.QueueSize.WARNING_SEND_QUEUE_SIZE);//预警数据集合
    public static final BlockingQueue<Integer> pointIdsQueue = new LinkedBlockingDeque<Integer>(ConCast.QueueSize.STOMP_QUEUE_SIZE);


    @Override
    public void run() {
        log.info("start decode data from ReceiveDataStorage");
        try {
            while(true){
                DataGram dataGram = ReceiveDataStorage.getInstance().receiveQueue.take();
                Integer pointId = dataGram.getHead().getPointId();
                //数据格式是否正确
                if (isErrorData(dataGram)){
                    // saveErrorData();
                    //todo cuowu shuju chuli
                    continue;
                }
                //将数据存放至发送stomp的队列中
                offerId2StompQueue(pointId);

                for (Param param:dataGram.getData()) {
                    WarningThreshold warningThreshold = WarningLevelUtil.getLevel(
                            dataGram.getHead().getPointId(),
                            param.getParameterId(),
                            Double.valueOf(param.getValue()));
                    if (warningThreshold.getLevelId() != -1){//预警数据
                        WarningData warningData = castDataGram2WarningData(dataGram,param,warningThreshold);
                        // 将预警数据存放入库队列中
                        offerWarningData2DBQueue(warningData);
                        // 将预警数据存放发送队列中
                        offerWarningData2SendQueue(warningData);
                    }
                    DeviceData data = castDataGram2DeviceData(dataGram,param,warningThreshold);

                    //将实时数据存放内存中，供Stomp调用
                    LoopQueue<DeviceData> loopQueue = loopQueueStorage(pointId,data);

                    //修改监测点的状态
                    changeObjectMonitoringStatus(loopQueue,dataGram);
                    //将DataGram转换成MonitoringData并放入入库队列中
                    MonitoringData monitoringData = castDataGram2MonitoringData(dataGram,param);
                    putMonitoringData2DBQueue(monitoringData);
                }
            }
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private void saveErrorData(DataGram dataGram,Param param) {
        MonitoringErrorData monitoringErrorData = new MonitoringErrorData();
        monitoringErrorData.setPointName(dataGram.getHead().getPointName());
        monitoringErrorData.setPointId(dataGram.getHead().getPointId());
        monitoringErrorData.setDeviceId(dataGram.getHead().getDeviceId());
        monitoringErrorData.setDeviceName(dataGram.getHead().getDeviceName());
        monitoringErrorData.setErrorValue(param.getValue());
        monitoringErrorData.setParameterId(param.getParameterId());
        monitoringErrorData.setParameterName(param.getParameterName());
        monitoringErrorData.setTime(dataGram.getHead().getTime());
        monitoringErrorData.setTypeId(dataGram.getHead().getTypeId());
        monitoringErrorData.setTypeName(dataGram.getHead().getTypeName());

        MonitoringErrorDataMapper monitoringErrorDataMapper = SpringApplicationContext.getBean(MonitoringErrorDataMapper.class);
        monitoringErrorDataMapper.insert(monitoringErrorData);
    }


    private MonitoringData castDataGram2MonitoringData(DataGram dataGram,Param param) {
        MonitoringData monitoringData = new MonitoringData();
        monitoringData.setPointName(dataGram.getHead().getPointName());//null
        monitoringData.setParameterId(param.getParameterId());
        monitoringData.setParameterName(param.getParameterName());// null
        monitoringData.setTime(dataGram.getHead().getTime());
        monitoringData.setValue(Double.valueOf(param.getValue()));
        monitoringData.setDeviceName(dataGram.getHead().getDeviceName());
        monitoringData.setDeviceId(dataGram.getHead().getDeviceId());
        monitoringData.setGroupId(dataGram.getHead().getTypeId());
        monitoringData.setGroupName(dataGram.getHead().getTypeName());//null
        monitoringData.setPointId(dataGram.getHead().getPointId());
        return monitoringData;
    }

    private void putMonitoringData2DBQueue(MonitoringData data) {
        try {
            monitoringDataQueue.put(data);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }

    private LoopQueue<DeviceData> loopQueueStorage(Integer pointId,DeviceData data) {
        String pointName = CacheDict.getPointNameByPointId(pointId);
        if (pointName == null){
            return null;
        }

        HashMap<String,LoopQueue<DeviceData>> parameterNames = new HashMap<>();
        if (pointMap.containsKey(pointName)){
            parameterNames= pointMap.get(pointName);
        }else{
            parameterNames = new HashMap<>();
            pointMap.put(pointName,parameterNames);
        }

        LoopQueue<DeviceData> deviceDataQueue =null;
        Set<String> parameterNameEns = CacheDict.getParameterNameEnsByPointId(pointId);//getParameterNameEnByParameterName(data.getId());
        for (String parameterNameEn:parameterNameEns) {
            if (parameterNames.containsKey(parameterNameEn)){
                deviceDataQueue = parameterNames.get(parameterNameEn);
            }else {
                deviceDataQueue = new LoopQueue<DeviceData>();
                parameterNames.put(parameterNameEn,deviceDataQueue);
            }

            deviceDataQueue.offer(data);
        }
        return deviceDataQueue;
    }

    private void changeObjectMonitoringStatus(LoopQueue<DeviceData> deviceDataQueue, DataGram dataGram) {
        DeviceData deviceData = null;
        try{
            deviceData = deviceDataQueue.getTail();
        }catch (Exception e){
            log.error("循环队列中数据为空");
        }
        if (deviceData != null){
            long preTime = 0;
            try {
                preTime = DateUtil.parse(deviceData.getTime(),DateUtil.PATTERN_YYYY_MM_DD_HHMMSS).getTime();
            } catch (Exception e) {
                log.error("parse time error {}",JsonUtil.obj2String(deviceData));
            }

            long newTime = DateUtil.parse(dataGram.getHead().getTime(),DateUtil.PATTERN_YYYY_MM_DD_HHMMSS).getTime();
            if (newTime - preTime > ObjectStatusMonitoring.timeout){//传感器设备异常已经恢复
                ObjectMonitoring objectMonitoring = new ObjectMonitoring();
                objectMonitoring.setStatus(0);
                objectMonitoring.setRecover(1);
                objectMonitoring.setMonitoringObject(dataGram.getHead().getPointName());
                objectMonitoring.setOperateTime(DateUtil.format(new Date(),DateUtil.PATTERN_YYYY_MM_DD_HHMM));
                objectMonitoring.setMonitoringType(1);
                ObjectMonitoringMapper objectMonitoringMapper = SpringApplicationContext.getBean(ObjectMonitoringMapper.class);
                objectMonitoringMapper.updateByObj(objectMonitoring);
            }
        }
    }

    private DeviceData castDataGram2DeviceData(DataGram dataGram, Param param,  WarningThreshold warningThreshold) {
        //实时推送数据
        DeviceData data = new DeviceData();
        data.setId(param.getParameterId());
        data.setName(param.getParameterName());
        data.setTime(dataGram.getHead().getTime());
        data.setValue(param.getValue());
        data.setWarnLevel(String.valueOf(warningThreshold.getLevelId()));
        data.setUtil(CacheDict.getUtilByParamterId(param.getParameterId()));
        return data;
    }

    private void offerWarningData2SendQueue(WarningData warningData) {
        Boolean putSendMsgQueue = null;
        try {
            putSendMsgQueue = sendMsgQueue.offer(warningData,1,TimeUnit.SECONDS);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
        if (!putSendMsgQueue){
            log.error("put send queue error data-> {}",JsonUtil.obj2String(warningData));
        }
    }

    private void offerWarningData2DBQueue(WarningData warningData) {
        Boolean putWarningQueue = null;
        try {
            putWarningQueue = warningQueue.offer(warningData,1,TimeUnit.SECONDS);
        } catch (InterruptedException e) {
//            e.printStackTrace();
        }
        if (!putWarningQueue){
            log.error("put warning queue error data-> {}",JsonUtil.obj2String(warningData));
        }
    }

    private WarningData castDataGram2WarningData(DataGram dataGram, Param param, WarningThreshold warningThreshold) {
        WarningData warningData =new WarningData();
        warningData.setDeviceId(dataGram.getHead().getDeviceId());
        warningData.setPointId(dataGram.getHead().getPointId());
        warningData.setParameterId(param.getParameterId());
        warningData.setLevelId(warningThreshold.getLevelId());

        warningData.setWarningValue(param.getValue());
        warningData.setState(0);
        warningData.setTypeId(param.getParameterId());//监测类型编号
        warningData.setWarningTime(dataGram.getHead().getTime());
        return warningData;
    }



    /**
     * 判断是否是异常数据，存在的问题：
     * 当数据转换失败之后，说明数据是异常数据，是否要进行异常数据存储待下一步明确
     * 转换正常的时候进行判断是否是一个峰值数据，详见
     *  @See {@link StorageTaskService#isExceptionData(Integer, Integer, Double)}
     * */
    private boolean isErrorData(DataGram dataGram) {
        if (dataGram == null){// 便于处理后续判断传感器状态
            return true;
        }
        for (Param param: dataGram.getData()){
            try {
                double value = Double.valueOf(param.getValue());
                if (isExceptionData(dataGram.getHead().getPointId(),param.getParameterId(),value)){
                    return true;
                }
            } catch (NumberFormatException e) {
                log.error("监测点 {} 出现了错误数据，数据内容为 {}", dataGram.getHead().getPointName(),JsonUtil.obj2String(dataGram));
                // TODO 错误数据入库
                return true;
            } catch (IllegalArgumentException e){
                log.error("执行KNN计算时出现了错误数据，数据内容为 {}", e.getMessage());
            }
        }
        return false;
    }

    /**
     * <p>判断数据是否是一个峰值数据，如果对应数据库无检测点，则说明是异常数据，直接返回，丢弃</p>
     * <p>否则将内存中暂存队列的数据取出形成double数组 + 当前点的数据代入KNN算法中进行计算，
     *  在KNN算法中，我们将当前数据与历史数据的距离均值大于历史数据两两之间的距离均值的
     *  {@link ConCast.KNN#THROULD_VALUE} 倍定义为异常数据，目前将异常数据直接丢弃，
     *  如要进行异常数据落库，则需要在return前一行进行入库处理，具体KNN算法实现,
     *  详见 @See {@link KNN#isExceptionData(List, Double)}</p>
     * */
    private boolean isExceptionData(Integer pointId,Integer parameterId,Double value) throws IllegalArgumentException {
        MonitoringPointMapper monitoringPointMapper = SpringApplicationContext.getBean(MonitoringPointMapper.class);

        MonitoringPoint monitoringPoint = monitoringPointMapper.selectByPrimaryKey(pointId);
        if (monitoringPoint == null){
            log.info("异常数据, 参数点 {} 无对应的参数编号 {} 其value为 {},", pointId, parameterId, value);
            return true;
        }
        String pointName = monitoringPoint.getPointName();
        String parameterName = CacheDict.getParameterNameByParameterId(parameterId);

        // todo 待斟酌 在此可以加上传感器状态监听处理
        if (! pointMap.containsKey(pointName)){
            return false;
        }
        if (!pointMap.get(pointName).containsKey(parameterName)){
            return false;
        }

        LoopQueue<DeviceData> dataLoopQueue = pointMap.get(pointName).get(parameterName);
        List<Double> oldData = new ArrayList<>();
        for (DeviceData data:dataLoopQueue.getData(10)) {
            oldData.add(Double.parseDouble(data.getValue()));
        }
        return KNN.isExceptionData(oldData,value);
    }



    private void offerId2StompQueue(Integer pointId) {
        boolean success = pointIdsQueue.offer(pointId);
        if (!success){
            log.info("将数据往stomp队列中存放时出现异常 {}",JsonUtil.obj2String(pointId));
        }
    }
}

//pg_dump.exe --host 39.100.44.189 --port 5532 --username "postgres" --password "123456"  --format custom --blobs --verbose --file "C:/postgres/qd9_portalCrane.backup" "qd9_portalCrane"
